package com.vlin.config.tuya;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.shade.org.apache.commons.lang.NullArgumentException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;

import com.vlin.api.mq.utils.ThreadPoolFactory;
import com.vlin.config.tuya.MessageHandlerConfig.MessageHandlerTask;
import com.vlin.spring.event.PulsarSpringEvent;

@Configuration
public class PulsarConfig implements ApplicationListener<PulsarSpringEvent>, ApplicationContextAware {

	private static final Logger logger = LoggerFactory.getLogger(PulsarConfig.class);


	
	@Autowired
	private ThreadPoolProperty poolProperty;


	private ApplicationContext context;

	@Override
	public void onApplicationEvent(PulsarSpringEvent event) {
		int num = poolProperty.getNum();

		if (num <= 0) {
			throw new NullArgumentException("Pulsar thread num is down to 0");
		}
		logger.info("--Pulsar thread pool start task--");
		for (int i = 0; i < num; i++) {
			Consumer<?> consumer = context.getBean(Consumer.class);
			MessageHandlerTask task = context.getBean(MessageHandlerTask.class, i, consumer);
			 ThreadPoolFactory.getCustomThreadPool().submit(task);
		}

	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.context = applicationContext;

	}

}
